package drds.binlog.parse;

import drds.binlog.common.AbstractLifeCycle;
import drds.binlog.common.Authors;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.common.position.LogIdentity;
import drds.binlog.common.position.LogPosition;
import drds.binlog.data_object.Entry;
import drds.binlog.data_object.EntryType;
import drds.binlog.data_object.Header;
import drds.binlog.database_driver.packets.GtidSet;
import drds.binlog.database_driver.packets.IGtidSet;
import drds.binlog.event_filter.EventFilter;
import drds.binlog.parse.exception.ParseException;
import drds.binlog.parse.exception.PositionNotFoundException;
import drds.binlog.parse.exception.TableIdNotFoundException;
import drds.binlog.parse.log_position_manager.LogPositionManager;
import drds.binlog.parse.mysql.AuthenticationInfo;
import drds.binlog.parse.mysql.MultiStageCoprocessor;
import drds.binlog.sink.IEntrySink;
import drds.binlog.sink.SinkException;
import drds.common.$;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.math.RandomUtils;
import org.slf4j.MDC;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicBoolean;


@Slf4j
@Author(name = Authors.LI_YANG)
public abstract class AbstractEventParser<Event> extends AbstractLifeCycle implements IEventParser<Event>
{


    public static final String destination_key = "destination";
    @Setter
    @Getter
    protected LogPositionManager logPositionManager = null;
    @Setter
    @Getter
    protected IEntrySink<List<Entry>> entrySink = null;
    @Setter
    @Getter
    protected EventFilter schemaNameAndTableNameRegexFilter = null;
    @Setter
    @Getter
    protected EventFilter blackSchemaNameAndTableNameRegexFilter = null;
    // 认证信息
    @Setter
    @Getter
    protected volatile AuthenticationInfo authenticationInfo;
    @Setter
    @Getter
    protected String destination;
    @Setter
    @Getter
    protected BinlogParser binlogParser = null;
    @Setter
    @Getter
    protected Thread parseThread = null;
    protected Thread.UncaughtExceptionHandler uncaughtExceptionHandler = new Thread.UncaughtExceptionHandler()
    {

        public void uncaughtException(Thread thread,
                                      Throwable e)
        {
            log.error("parse events has an error", e);
        }
    };
    @Setter
    @Getter
    protected EntryListTransaction entryListTransaction;
    @Setter
    @Getter
    protected int transactionSize = 1024;
    @Setter
    @Getter
    protected AtomicBoolean needTransactionPosition = new AtomicBoolean(false);
    @Setter
    @Getter
    protected long lastEntryTime = 0L;
    @Setter
    @Getter
    protected volatile boolean detectingEnable = true;                                    // 是否开启心跳检查
    @Setter
    @Getter
    protected Integer detectingIntervalInSeconds = 3;                                       // 检测频率
    @Setter
    @Getter
    protected volatile Timer timer;
    @Setter
    @Getter
    protected TimerTask heartBeatTimerTask;
    @Setter
    @Getter
    protected Throwable exception = null;
    @Setter
    @Getter
    protected boolean isGTIDMode = false;                                   // 是否是GTID模式
    @Setter
    @Getter
    protected boolean parallel = true;                                    // 是否开启并行解析模式
    @Setter
    @Getter
    protected Integer parallelThreadSize = Runtime.getRuntime().availableProcessors() * 60 / 100;     // 60%的能力跑解析,剩余部分处理网络
    @Setter
    @Getter
    protected int parallelBufferSize = 256;                                     // 必须为2的幂
    @Setter
    @Getter
    protected IMultiStageCoprocessor multiStageCoprocessor;
    @Setter
    @Getter
    protected ParserExceptionHandler parserExceptionHandler;
    protected long serverId;

    public AbstractEventParser()
    {
        // 初始化一下
        entryListTransaction = new EntryListTransaction(new EntryListTransactionFlush()
        {

            public void flush(List<Entry> entryList) throws InterruptedException
            {
                boolean successed = sink(entryList);
                if (!running)
                {
                    return;
                }

                if (!successed)
                {
                    throw new ParseException("consume failed!");
                }

                LogPosition logPosition = buildLastTransactionPosition(entryList);
                if (logPosition != null)
                { // 可能position为空
                    logPositionManager.persistLogPosition(AbstractEventParser.this.destination, logPosition);
                }
            }
        });
    }

    protected abstract BinlogParser buildParser();

    protected abstract IConnection buildConnection();

    protected abstract IMultiStageCoprocessor buildMultiStageCoprocessor();

    protected abstract EntryPosition findStartPosition(IConnection connection) throws IOException;

    protected void preDump(IConnection connection)
    {
    }

    protected boolean processTableMetaData(EntryPosition entryPosition)
    {
        return true;
    }

    protected void afterDump(IConnection connection)
    {
    }

    public void start()
    {
        super.start();
        MDC.put(destination_key, destination);
        // 配置transaction bytes
        // 初始化缓冲队列
        entryListTransaction.setBufferSize(transactionSize);// 设置buffer大小
        entryListTransaction.start();
        // 构造bin log parser
        binlogParser = buildParser();// 初始化一下BinLogParser
        binlogParser.start();
        // 启动工作线程
        parseThread = new Thread(new Runnable()
        {

            public void run()
            {
                MDC.put("destination", String.valueOf(destination));
                IConnection connection = null;
                while (running)
                {
                    try
                    {
                        // 开始执行replication
                        // 1. 构造连接
                        connection = buildConnection();
                        log.info("connection:" + connection);
                        // 2. 启动一个心跳线程
                        startHeartBeatTimeTask(connection);

                        // 3. 执行dump前的准备工作
                        preDump(connection);

                        connection.connect();// 链接

                        long queryServerId = connection.queryServerId();
                        if (queryServerId != 0)
                        {
                            serverId = queryServerId;
                        }
                        // 4. 获取最后的位置信息
                        EntryPosition entryPosition = findStartPosition(connection);
                        final EntryPosition startPosition = entryPosition;
                        if (startPosition == null)
                        {
                            throw new PositionNotFoundException("can't find start setIndexAsOriginIndexAndOffset$originIndex for " + destination);
                        }

                        if (!processTableMetaData(startPosition))
                        {
                            throw new ParseException("can't find unsettled table metadata for " + destination
                                    + " with setIndexAsOriginIndexAndOffset$originIndex : " + startPosition);
                        }
                        log.warn("find start setIndexAsOriginIndexAndOffset$originIndex : {}", startPosition.toString());
                        // 重新链接，因为在找position过程中可能有状态，需要断开后重建
                        connection.reconnect();

                        final SinkFunction sinkFunction = new SinkFunction<Event>()
                        {

                            private LogPosition lastPosition;

                            public boolean sink(Event event)
                            {
                                try
                                {
                                    Entry entry = parse(event, false);

                                    if (!running)
                                    {
                                        return false;
                                    }

                                    if (entry != null)
                                    {
                                        exception = null; // 有正常数据流过，清空exception
                                        entryListTransaction.add(entry);
                                        // 记录一下对应的positions
                                        this.lastPosition = buildLastPosition(entry);
                                        // 记录一下最后一次有数据的时间
                                        lastEntryTime = System.currentTimeMillis();
                                    }
                                    return running;
                                } catch (TableIdNotFoundException e)
                                {
                                    throw e;
                                } catch (Throwable e)
                                {
                                    if (e.getCause() instanceof TableIdNotFoundException)
                                    {
                                        throw (TableIdNotFoundException) e.getCause();
                                    }
                                    // 记录一下，出错的位点信息
                                    processSinkError(e,
                                            this.lastPosition,
                                            startPosition.getJournalName(),
                                            startPosition.getPosition());
                                    throw new ParseException(e); // 继续抛出异常，让上层统一感知
                                }
                            }

                        };

                        // 4. 开始dump数据
                        if (parallel)
                        {
                            // build stage processor
                            multiStageCoprocessor = buildMultiStageCoprocessor();
                            if (isGTIDMode())
                            {
                                // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据
                                IGtidSet gtidSet = GtidSet.parse(startPosition.getGtid());
                                ((MultiStageCoprocessor) multiStageCoprocessor).setGtidSet(gtidSet);
                                multiStageCoprocessor.start();
                                connection.dump(gtidSet, multiStageCoprocessor);
                            } else
                            {
                                multiStageCoprocessor.start();
                                if ($.isNullOrEmpty(startPosition.getJournalName())
                                        && startPosition.getTimestamp() != null)
                                {
                                    connection.dump(startPosition.getTimestamp(), multiStageCoprocessor);
                                } else
                                {
                                    connection.dump(startPosition.getJournalName(),
                                            startPosition.getPosition(),
                                            multiStageCoprocessor);
                                }
                            }
                        } else
                        {
                            if (isGTIDMode())
                            {
                                // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据
                                connection.dump(GtidSet.parse(startPosition.getGtid()), sinkFunction);
                            } else
                            {
                                if ($.isNullOrEmpty(startPosition.getJournalName())
                                        && startPosition.getTimestamp() != null)
                                {
                                    connection.dump(startPosition.getTimestamp(), sinkFunction);
                                } else
                                {
                                    connection.dump(startPosition.getJournalName(),
                                            startPosition.getPosition(),
                                            sinkFunction);
                                }
                            }
                        }
                    } catch (TableIdNotFoundException e)
                    {
                        exception = e;
                        // 特殊处理TableIdNotFound异常,出现这样的异常，一种可能就是起始的position是一个事务当中，导致tablemap
                        // Event时间没解析过
                        needTransactionPosition.compareAndSet(false, true);
                        log.error(String.format("dump address %s has an error, retrying. caused by ",
                                authenticationInfo.getInetSocketAddress().toString()), e);
                    } catch (Throwable e)
                    {
                        processDumpError(e);
                        exception = e;
                        if (!running)
                        {
                            if (!(e instanceof java.nio.channels.ClosedByInterruptException || e.getCause() instanceof java.nio.channels.ClosedByInterruptException))
                            {
                                throw new ParseException(String.format("dump address %s has an error, retrying. ",
                                        authenticationInfo.getInetSocketAddress().toString()), e);
                            }
                        } else
                        {
                            log.error(String.format("dump address %s has an error, retrying. caused by ",
                                    authenticationInfo.getInetSocketAddress().toString()), e);
                            //sendAlarm(destination, ExceptionUtils.getFullStackTrace(e));
                        }
                        if (parserExceptionHandler != null)
                        {
                            parserExceptionHandler.handle(e);
                        }
                    } finally
                    {
                        // 重新置为中断状态
                        Thread.interrupted();
                        // 关闭一下链接
                        afterDump(connection);
                        try
                        {
                            if (connection != null)
                            {
                                connection.disconnect();
                            }
                        } catch (IOException e1)
                        {
                            if (!running)
                            {
                                throw new ParseException(String.format("disconnect address %s has an error, retrying. ",
                                        authenticationInfo.getInetSocketAddress().toString()),
                                        e1);
                            } else
                            {
                                log.error("disconnect address {} has an error, retrying., caused by ",
                                        authenticationInfo.getInetSocketAddress().toString(),
                                        e1);
                            }
                        }
                    }
                    // 出异常了，退出sink消费，释放一下状态
                    entrySink.interrupt();
                    entryListTransaction.reset();// 重置一下缓冲队列，重新记录数据
                    binlogParser.reset();// 重新置位
                    if (multiStageCoprocessor != null)
                    {
                        // 处理 RejectedExecutionException
                        try
                        {
                            multiStageCoprocessor.reset();
                        } catch (Throwable t)
                        {
                            log.debug("multi processor rejected:", t);
                        }
                    }

                    if (running)
                    {
                        // sleep一段时间再进行重试
                        try
                        {
                            Thread.sleep(10000 + RandomUtils.nextInt(10000));
                        } catch (InterruptedException e)
                        {
                        }
                    }
                }
                MDC.remove("destination");
            }
        });

        parseThread.setUncaughtExceptionHandler(uncaughtExceptionHandler);
        parseThread.setName(String.format("destination = %s , address = %s , EventParser",
                destination,
                authenticationInfo == null ? null : authenticationInfo.getInetSocketAddress()));
        parseThread.start();
    }

    public void stop()
    {
        super.stop();

        stopHeartBeatTimerTask(); // 先停止心跳
        parseThread.interrupt(); // 尝试中断
        entrySink.interrupt();

        if (multiStageCoprocessor != null && multiStageCoprocessor.isRunning())
        {
            multiStageCoprocessor.stop();
        }

        try
        {
            parseThread.join();// 等待其结束
        } catch (InterruptedException e)
        {
            // ignore
        }

        if (binlogParser.isRunning())
        {
            binlogParser.stop();
        }
        if (entryListTransaction.isRunning())
        {
            entryListTransaction.stop();
        }
    }

    protected boolean sink(List<Entry> entryList) throws SinkException, InterruptedException
    {
        return entrySink.sink(entryList, (authenticationInfo == null) ? null : authenticationInfo.getInetSocketAddress(), destination);
    }

    protected Entry parse(Event event, boolean isSeek) throws Exception
    {
        return binlogParser.parse(event, isSeek);
    }


    protected LogPosition buildLastTransactionPosition(List<Entry> entryList)
    {
        for (int i = entryList.size() - 1; i > 0; i--)
        {
            Entry entry = entryList.get(i);
            if (entry.getEntryType() == EntryType.transaction_end)
            {   // 尽量记录一个事务做为position
                return buildLastPosition(entry);
            }
        }

        return null;
    }

    protected LogPosition buildLastPosition(Entry entry)
    {
        LogPosition logPosition = new LogPosition();
        //
        EntryPosition entryPosition = new EntryPosition();
        entryPosition.setJournalName(entry.getHeader().getLogFileName());
        entryPosition.setPosition(entry.getHeader().getLogFileOffset());
        entryPosition.setTimestamp(entry.getHeader().getExecuteDateTimeLong());
        // add serverId at 2016-06-28
        entryPosition.setServerId(entry.getHeader().getServerId());
        // set gtid
        entryPosition.setGtid(entry.getHeader().getGtid());
        //
        logPosition.setEntryPosition(entryPosition);
        //
        LogIdentity logIdentity = new LogIdentity(authenticationInfo.getInetSocketAddress(), -1L);
        logPosition.setLogIdentity(logIdentity);
        return logPosition;
    }

    protected void processSinkError(Throwable e, LogPosition lastPosition, String startBinlogFileName, Long startBinlogFilePosition)
    {
        if (lastPosition != null)
        {
            log.warn(String.format("ERROR ## parse this event has an error , last setIndexAsOriginIndexAndOffset$originIndex : [%s]",
                    lastPosition.getEntryPosition()),
                    e);
        } else
        {
            log.warn(String.format("ERROR ## parse this event has an error , last setIndexAsOriginIndexAndOffset$originIndex : [%s,%s]",
                    startBinlogFileName,
                    startBinlogFilePosition), e);
        }
    }

    protected void processDumpError(Throwable e)
    {
        // do nothing
    }

    protected void startHeartBeatTimeTask(IConnection connection)
    {
        lastEntryTime = 0L; // 初始化
        if (timer == null)
        {
            String name = String.format("destination = %s , address = %s , HeartBeatTimeTask",//
                    destination,//
                    authenticationInfo == null ? null : authenticationInfo.getInetSocketAddress().toString());//
            synchronized (AbstractEventParser.class)
            {
                if (timer == null)
                {
                    timer = new Timer(name, true);
                }
            }
        }

        if (heartBeatTimerTask == null)
        {   //避免重复创建heartbeat线程
            heartBeatTimerTask = buildHeartBeatTimeTask(connection);
            Integer interval = detectingIntervalInSeconds;
            timer.schedule(heartBeatTimerTask, interval * 1000L, interval * 1000L);
            log.info("start heart beat.... ");
        }
    }

    protected TimerTask buildHeartBeatTimeTask(IConnection connection)
    {
        return new TimerTask()
        {

            public void run()
            {
                try
                {
                    if (exception == null || lastEntryTime > 0)
                    {
                        // 如果未出现异常，或者有第一条正常数据
                        long now = System.currentTimeMillis();
                        long inteval = (now - lastEntryTime) / 1000;
                        if (inteval >= detectingIntervalInSeconds)
                        {
                            Header header = new Header();
                            header.setExecuteDateTimeLong(now);
                            Entry entry = new Entry();
                            entry.setHeader(header);
                            entry.setEntryType(EntryType.heartbeat);
                            // 提交到sink中,会在sink中进行忽略,目前不会提交到store中.
                            sink(Arrays.asList(entry));
                        }
                    }

                } catch (Throwable e)
                {
                    log.warn("heartBeat run failed ", e);
                }
            }

        };
    }

    protected void stopHeartBeatTimerTask()
    {
        lastEntryTime = 0L; // 初始化
        if (timer != null)
        {
            timer.cancel();
            timer = null;
        }
        heartBeatTimerTask = null;
    }


}
